package com.lx.demo.sync.executor;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;

import  com.lx.demo.common.PageResult;
import com.lx.demo.cache.CacheUtil;
import com.lx.demo.common.WrapperResponse;
import com.lx.demo.constans.BaseConstant;
import com.lx.demo.constans.ExtDataEsInxConstant;
import com.lx.demo.sync.base.ExtBaseDTO;
import com.lx.demo.sync.base.ExtFetchDataRequest;
import com.lx.demo.sync.base.ExtPage;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 外部数据同步抽象类
 * 抽取公共逻辑
 * 1.比如获取拉取数据的时间戳或者最大版本号
 * 2.任务完成后记录日志
 * 优化点  动态设置是否异步入库  每次同步数据量,还要防止重复执行
 * @author wcy
 */
public abstract class ExtSyncAbstract {

    private static final Logger log = LoggerFactory.getLogger(ExtSyncAbstract.class);
    @Autowired
    private CacheUtil cacheUtil;

    public ExtSyncAbstract() {
    }

    /**
     * 子类实现,获取业务类唯一标识
     * @return
     */
    protected abstract String getIdxName();

    /**
     * 获取数据
     * @param request
     * @return
     */
    protected abstract ExtPage fetchSourceData(ExtFetchDataRequest request);

    /**
     * 同步到数据库 由各业务类实现具体逻辑
     * @param esSyncModelList
     * @throws Exception
     */
    protected abstract void bulkToDb(List<?> esSyncModelList);

    /**
     * 数据同步 --待优化  动态设置最大拉取数量
     * @return
     */
    private final  long sync(){

        long count = 0;
        ExtFetchDataRequest request =
                new ExtFetchDataRequest(getLastUpdateTime(), BaseConstant.START_PAGE_NUM, BaseConstant.EXT_PULL_BATCH_SIZE);

        // 首次查询, 并获取分页信息
        ExtPage extPage = fetchSourceData(request);
        if (Objects.isNull(extPage)) {
            return 0L;
        }
        List<? extends ExtBaseDTO> docs = extPage.getRecords();
        log.info("  >>> 同步 [{}], 总条数 :: {} ", getIdxName(), extPage.getTotalPages());
        bulkAndUpdateTime(docs);
        count += docs.size();
        while (extPage.getPageNumber() < extPage.getTotalPages()) {
            request.setPageNum(extPage.getPageNumber() + 1);
            // 一次同步只更新分页信息, 不同步更新时间
            log.info("  >>> 拉取外部数据 :: [{}] 开始执行, 当前页 :: [{}], 当前拉取条数 :: [{}], 满足记录总行数 :: [{}], 总页数 :: [{}] ",
                    getIdxName(), extPage.getPageNumber(), extPage.getPageSize(), extPage.getRecordCount(), extPage.getTotalPages());
            extPage = fetchSourceData(request);
            if (Objects.isNull(extPage)) {
                log.warn("  >>> 拉取外部数据为空 :: [{}] ",
                        getIdxName());
                break;
            }
            docs = extPage.getRecords();
            bulkAndUpdateTime(docs);
            count += docs.size();
        }

        return count;
    }

    public final void pullExternalDataAndSync(String param) {
        String index = getIdxName();
        log.info("  >>>  开始同步数据, 类型名称 :: [{}], 缓存 key :: [{}]", index, getCacheKey());
        try {
            if (Objects.equals(BaseConstant.FULL_UPDATE_FLAG, param)) {
                cacheUtil.del(getCacheKey());
                log.info("  >>> 全量同步 :: [{}]", index);
            }
            log.info("  >>>  获取的最后更新时间 :: [{}]", getFormatLastUpdateTime());
            long count = sync();
            log.info("  >>>  共处理 [{}] 条数据, 记录的最后更新时间 :: [{}]", count, getFormatLastUpdateTime());
        } catch (Exception e) {
            log.error(String.format("  >>>  同步外部数据异常, 类型名称 :: [%s]", index), e);
        }
    }

    /**
     * 获取上次任务最后的更新时间
     * @return
     */
    private Date getLastUpdateTime() {
        Date lastUpdateTime = cacheUtil.get(getCacheKey(), Date.class);
        return lastUpdateTime == null ? DateUtil.parse("1970-01-01") : lastUpdateTime;
    }

    /**
     * 获取记录最大更新数据缓存key
     * @return
     */
    private String getCacheKey() {
        StringBuilder cacheKey =
                new StringBuilder(ExtDataEsInxConstant.REDIS_EXTERNAL_ES_PREFIX)
                        .append(getIdxName());
        return cacheKey.toString();
    }

    /**
     * 存储数据至ES或DB, 累计操作数据数量, 更新最后更新时间
     *
     * @param docs
     * @return
     */
    private void bulkAndUpdateTime(List<? extends ExtBaseDTO> docs) {
        SyncExecutor executor = ExtConfiguration.newSyncExecutor(this);
        executor.executor(docs);
        saveLastUpdateTime(docs);
    }

    /**
     * 保存最后更新时间
     * @param docs
     */
    private void saveLastUpdateTime(List<? extends ExtBaseDTO> docs) {
        if (docs.size() == 0) {
            return;
        }
        String index = getCacheKey();
        Date newestTime = docs.get(docs.size() - 1).getUpdtTime();
        log.info("  >>>  同步数据 :: [{}], 记录最后更新时间 :: [{}]", index, DateUtil.format(newestTime, DatePattern.NORM_DATETIME_PATTERN));
        cacheUtil.set(getCacheKey(), newestTime, 365, TimeUnit.DAYS);
    }

    /**
     * 获取格式化最后更新时间
     *
     * @return
     */
    private String getFormatLastUpdateTime() {
        return DateUtil.format(getLastUpdateTime(), DatePattern.NORM_DATETIME_PATTERN);
    }


    protected <T> boolean isSuccessAndData(WrapperResponse<PageResult<T>> wrapperResponse) {
        if (Objects.isNull(wrapperResponse)
                || !Objects.equals(wrapperResponse.getCode(), BaseConstant.SUCCESS_FLAG)
                || Objects.isNull(wrapperResponse.getData())
                || CollectionUtils.isEmpty(wrapperResponse.getData().getData())) {
            return false;
        }
        return true;
    }

}
